package com.chenxu.gmall.realtime.app.dws;
import com.chenxu.gmall.realtime.app.func.KeywordProductC2RUDTF;
import com.chenxu.gmall.realtime.app.func.KeywordUDTF;
import com.chenxu.gmall.realtime.bean.KeywordStats;
import com.chenxu.gmall.realtime.utils.ClickHouseUtil;
import com.chenxu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Date: 2021/07/20
 * Desc:  从商品统计中（dws_product_stats）获取关键词
 * 自定义 UDTF 函数实现点击次数、订单次数、添加购物次数的统计；存入clickhouse中；
 */
/*
        整体测试
        ➢ 启动 ZK、Kafka、logger.sh、ClickHouse、Redis、HDFS、Hbase、Maxwell
        ➢ 运行 BaseLogApp
        ➢ 运行 BaseDBApp
        ➢ 运行 OrderWideApp
        ➢ 运行 PaymentWideApp
        ➢ 运行 ProductsStatsApp
        ➢ 运行 KeywordStats4ProductApp
        ➢ 运行 rt_applog 目录下的 jar 包
        ➢ 运行 rt_dblog 目录下的 jar 包
        ➢ 查看控制台输出
        ➢ 查看 ClickHouse 中 products_stats_0709 表数据

        keywordStatsProductDataStream数据类型为：
        //分别对应点击、订单和购物车；
        1> KeywordStats(keyword=10x, ct=2, source=CLICK, stt=2021-07-20 13:29:10, edt=2021-07-20 13:29:20, ts=1626758982000)
        1> KeywordStats(keyword=redmi, ct=12, source=ORDER, stt=2021-07-20 13:28:50, edt=2021-07-20 13:29:00, ts=1626758978000)
        1> KeywordStats(keyword=清, ct=178, source=CART, stt=2021-07-19 21:58:10, edt=2021-07-19 21:58:20, ts=1626758976000)

        clickhouse中数据为：
        ┌─────────────────stt─┬─────────────────edt─┬─keyword───┬─source─┬─ct─┬────────────ts─┐
        │ 2021-07-20 13:27:40 │ 2021-07-20 13:27:50 │ 10        │ CLICK  │  1 │ 1626758976000 │
        │ 2021-07-20 13:27:40 │ 2021-07-20 13:27:50 │ 10x       │ CLICK  │  1 │ 1626758977000 │
        │ 2021-07-20 13:27:40 │ 2021-07-20 13:27:50 │ redmi     │ CLICK  │  1 │ 1626758977000 │
        │ 2021-07-20 13:27:40 │ 2021-07-20 13:27:50 │ 小米      │ CLICK  │  1 │ 1626758976000 │
        │ 2021-07-20 13:27:50 │ 2021-07-20 13:28:00 │ 10        │ CLICK  │  2 │ 1626758977000 │
        │ 2021-07-20 13:27:50 │ 2021-07-20 13:28:00 │ 10        │ ORDER  │  9 │ 1626758977000 │

         */
public class KeywordStats4ProductApp {
    public static void main(String[] args) throws Exception {
        //TODO 0.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);
        /*
        //CK相关设置
        env.enableCheckpointing(5000, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        StateBackend fsStateBackend = new FsStateBackend(
                "hdfs://hadoop201:8020/gmall/flink/checkpoint/ProvinceStatsSqlApp");
        env.setStateBackend(fsStateBackend);
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */
        //TODO 1.定义Table流环境
        EnvironmentSettings settings = EnvironmentSettings
            .newInstance()
            .inStreamingMode()
            .build();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        //TODO 2.注册自定义函数
        tableEnv.createTemporarySystemFunction("ik_analyze",  KeywordUDTF.class);
        tableEnv.createTemporarySystemFunction("keywordProductC2R",  KeywordProductC2RUDTF.class);

        //TODO 3.将数据源定义为动态表
        String groupId = "keyword_stats_app";
        String productStatsSourceTopic ="dws_product_stats";

        //数据来源于kafka主题dws_product_stats；
        //数据类型大致为：
        //>>>>:1> ProductStats(stt=2021-07-19 16:43:00, edt=2021-07-19 16:43:10, sku_id=13,
        // sku_name=华为 HUAWEI P40 麒麟990 5G SoC芯片 5000万超感知徕卡三摄 30倍数字变焦 6GB+128GB亮黑色全网通5G手机,
        // sku_price=4188, spu_id=4, spu_name=HUAWEI P40, tm_id=3, tm_name=华为, category3_id=61, category3_name=手机,
        // display_ct=0, click_ct=0, favor_ct=0, cart_ct=211, order_sku_num=0, order_amount=0, order_ct=0,
        // payment_amount=0, paid_order_ct=0, refund_order_ct=0, refund_amount=0, comment_ct=0, good_comment_ct=0,
        // orderIdSet=[], paidOrderIdSet=[], refundOrderIdSet=[], ts=1626684201676)
        tableEnv.executeSql("CREATE TABLE product_stats (spu_name STRING, " +
            "click_ct BIGINT," +
            "cart_ct BIGINT," +
            "order_ct BIGINT ," +
            "stt STRING,edt STRING ) " +
            "  WITH ("+ MyKafkaUtil.getKafkaDDL(productStatsSourceTopic,groupId)+")");

        //主题dws_product_stats中数据包含内容：click_ct=0, favor_ct=0, cart_ct=211, order_sku_num=0, order_amount=0, order_ct=0；
        //TODO 6.聚合计数
        //UNIX_TIMESTAMP()：以秒为单位获取当前的Unix时间戳；
        //因为脚本数据生成时间也是按照当前时间生成的；
        //as T2(ct,source)是因为KeywordProductC2RUDTF的返回结果为：ROW<ct BIGINT,source STRING>；
        Table keywordStatsProduct = tableEnv.sqlQuery("select keyword,ct,source, " +
            "DATE_FORMAT(stt,'yyyy-MM-dd HH:mm:ss')  stt," +
            "DATE_FORMAT(edt,'yyyy-MM-dd HH:mm:ss') as edt, " +
            "UNIX_TIMESTAMP()*1000 ts from product_stats  , " +
            "LATERAL TABLE(ik_analyze(spu_name)) as T(keyword) ," +
            "LATERAL TABLE(keywordProductC2R( click_ct ,cart_ct,order_ct)) as T2(ct,source)");

        //TODO 7.转换为数据流
        DataStream<KeywordStats> keywordStatsProductDataStream =
            tableEnv.<KeywordStats>toAppendStream(keywordStatsProduct, KeywordStats.class);

        keywordStatsProductDataStream.print();
        //TODO 8.写入到ClickHouse
        keywordStatsProductDataStream.addSink(
            ClickHouseUtil.<KeywordStats>getJdbcSink(
                "insert into keyword_stats_0709(keyword,ct,source,stt,edt,ts)  " +
                    "values(?,?,?,?,?,?)"));



        env.execute();
    }
}
